package com.atguigu.tingshu.common.service;

import com.atguigu.tingshu.common.constant.KafkaConstant;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;


@Slf4j
@Service
public class KafkaService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaService.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;


    /**
     * 生产者：发送消息到Kafka话题
     *
     * @param topic 话题名称
     * @param data  消息内容
     */
    public void sendMessage(String topic, String data) {
        kafkaTemplate.send(topic, null, data);
    }

    /**
     * 生产者：发送消息到Kafka话题
     *
     * @param topic 话题名称
     * @param key   消息Key
     * @param data  消息内容
     */
    public void sendMessage(String topic, String key, String data) {
        //1.调用模板对象方法发送消息
        CompletableFuture completableFuture = kafkaTemplate.send(topic, key, data);
        //2.获取发送消息结果
        completableFuture.thenAcceptAsync(result -> {
            logger.info("[发消息工具方法]发送消息成功：{}", result);
        }).exceptionallyAsync(e -> {
            logger.error("[发消息工具方法]发送消息失败：{}", e);
            return null;
        });
    }



}
